#include "Process_poll.h"
std::shared_ptr<process_poll> process_poll::_ppoll_instance = nullptr;
static int sig_pipe[2];
std::mutex mtx;
//在这里规定 sig_pipe[1]
static int SetNonblockFd(int fd)
{
    assert(fd>0);
    return fcntl(fd,F_SETFL,O_NONBLOCK | fcntl(fd,F_GETFL));
}

static void sig_handler(int sig)       //通过 信号管道 发送 信号
{
    int errno_1 = errno;
    int msg = sig;
    //将信号的数值转变为宏的名字 在接受信号时就可以用宏名来进行筛选,这样在后续的信号处理时更好判断、更明确信号的含义
    //每次传一个字节 即使宏名很长 但是在所有进程当中 sig_pipe[1] 已经设置为非阻塞状态
    send(sig_pipe[1],(char*)&msg,1,0);  
    //进程被信号中断时,进行回射,触发二次自定义信号处理

    errno = errno_1;
}
static void add_sig(int sig,void(handler)(int) ,bool restart) 
{
    struct sigaction sigact;
    memset(&sigact,0,sizeof(sigact));
    sigact.sa_handler = handler;
    if(restart)
    {
        sigact.sa_flags |=SA_RESTART;       //在完成信号处理函数时,重新启动系统调用？？
    }
    sigfillset(&sigact.sa_mask);        //将所有的信号设置为 在信号处理函数运行时 进行等待/搁置
    assert(sigaction(sig,&sigact,NULL)>=0);
}


process_poll::process_poll(int port,int processMax,int timeoutMS,
                            int trigMode,bool Optlinger,bool Openlog,int loglevel,int logcapacity)
                            :port_(port),_NOW_PROCES_NUMBER(processMax),_proc_index(-1),
                            timeoutMS_(timeoutMS),_proc_status(false),_proc_atribute(nullptr),
                            ListenFd_(-1),OptLinger_(Optlinger),epoller_(new Epoller())
{
    assert( (processMax > 0) && (processMax <= __POLL_MAX_VOLUME));
    InitEventMode(trigMode);
    //初始化描述进程结构体数组
    _proc_atribute=new process[processMax];
    assert( InitSock()==true );

    for(int i=0;i<processMax;++i)
    {
        //为每个进程添加通信通道
        int stat = socketpair(PF_UNIX,SOCK_STREAM,0,_proc_atribute[i].pipe);
        assert(stat ==0);
        _proc_atribute[i].self_id = fork();
        assert(_proc_atribute[i].self_id>=0 );
        if(_proc_atribute[i].self_id == 0) 
        {
            //子进程 关闭子进程的 fd[1] 写管道？ 虽然fd[0]、fd[1] 两者都可以都有读写功能
            close(_proc_atribute[i].pipe[0]);
            SetNonblockFd(_proc_atribute[i].pipe[1]);
            _proc_index = i;
            break;
        }
        if(_proc_atribute[i].self_id>0)        
        {
            //父进程
            close(_proc_atribute[i].pipe[1]);
            SetNonblockFd(_proc_atribute[i].pipe[0]);
            continue;
        }
    }
}

void process_poll::setup_sig_pipe()
{
    int ret = socketpair(PF_UNIX,SOCK_STREAM,0,sig_pipe);
    assert(ret !=-1);

    SetNonblockFd(sig_pipe[1]);             
    epoller_->AddFd_(sig_pipe[0],pipe_event);
    //SetNonblockFd(sig_pipe[0]);

    add_sig(SIGINT,sig_handler,true);
    add_sig(SIGCHLD,sig_handler,true);
    add_sig(SIGTERM,sig_handler,true);
    add_sig(SIGPIPE,SIG_IGN);
}

void process_poll::run()
{
    if(_proc_index==-1)
    {
        run_parent();
    }
    else
    {
        run_child();
        return;
    }
}

bool process_poll::InitSock()
{
    int status;
    struct sockaddr_in Client_sockaddr;
    if(port_ >= 65535 || port_ < 1024)
    {
        return false;
    }

    //create sock
    status = socket(AF_INET,SOCK_STREAM,0);
    if(status < 0)
    {
        LOG_ERRNO("sock(AF_INET,SOCK_STREAM,0) Failed");
        return false;
    }
    ListenFd_ = status;

    //set socke option
    //优雅退出
    struct linger optlinger;
    bzero(&optlinger,sizeof(optlinger));
    if(OptLinger_)
    {
        optlinger.l_onoff = 3;
        //延遲1s彻底关闭sock,如果数据还未发送完,则直接丢弃数据
        optlinger.l_linger = 1; 
    }
    status = setsockopt(ListenFd_,SOL_SOCKET,SO_LINGER,&optlinger,sizeof(optlinger));
    //端口复用 干掉TIME_WAIT状态
    int optval = 1;
    status = setsockopt(ListenFd_,SOL_SOCKET,SO_REUSEPORT,&optval,sizeof(optval));
    if(status <0 )
    {
        LOG_ERRNO("Setsockopt Failed");
        close(ListenFd_);
        return false;
    }


    //bind
    bzero(&Client_sockaddr,sizeof(Client_sockaddr));
    Client_sockaddr.sin_family = AF_INET;
    Client_sockaddr.sin_port = htons(port_);
    Client_sockaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    status = bind(ListenFd_,(struct sockaddr*)&Client_sockaddr,sizeof(Client_sockaddr));
    if(status <0 )
    {
        LOG_ERRNO("Sock Bind Failed");
        close(ListenFd_);
        return false;
    }

    //listen
    status = listen(ListenFd_,1024);
    if(status <0 )
    {
        LOG_ERRNO("sock listen Failed");
        close(ListenFd_);
        return false;
    }
    //设置 监听符的属性,只监听读事件
    status = epoller_->AddFd_(ListenFd_,listen_event|EPOLLIN);
    if(status == 0)
    {
        LOG_ERRNO("epoller AddFd_ Failed");
        close(ListenFd_);
        return false;
    }

    //在边沿触发模式下,对监听sock的属性必须为非阻塞模式,
    //否则主进程会一直阻塞在accept函数上 
    status = SetNonblockFd(ListenFd_);
    LOG_INFO("sock port: %d",port_);
    return true;
}
void process_poll::DealAccept()
{
    struct sockaddr_in Client_sockaddr;
    socklen_t addr_len = sizeof(Client_sockaddr);
    do
    {
        int status = accept(ListenFd_,(struct sockaddr*)&Client_sockaddr,&addr_len);
        if(status <=0)  //EWOULDBLOCK 、EAGIN
        {   return; }
        else if(Httpconn::usercount_>= MAX_FD)
        {
            SendError(status,"Server is full");
            LOG_ERRNO("Client Is Full");
            return;
        }
        AddClient(status,Client_sockaddr);
        
    } while (listen_event & EPOLLET);
}
void process_poll::AddClient(int fd,sockaddr_in addr)
{
    assert(fd > 0);
    user_[fd].init(fd,addr);
    if(timeoutMS_ > 0)
    {
        Timer_->add(fd,timeoutMS_,std::bind(&process_poll::CloseCoon,this,&user_[fd]));
    }
    epoller_->AddFd_(fd,coonfd_event|EPOLLIN);
    SetNonblockFd(fd);
    //printf("Client[%d] linked in!\n",user_[fd].getFd());
}
void process_poll::InitEventMode(int Mode)
{
    pipe_event = EPOLLIN | EPOLLET;
    listen_event = EPOLLRDHUP;                  //检测 主进程 监听sock状态
    coonfd_event = EPOLLIN | EPOLLONESHOT;      //在Reactor模式下，一次完整的事务只由单个线程完成
    switch(Mode)
    {
        case 0:
        {
            break;
        }
        case 1:
        {
            listen_event |= EPOLLET;
            break;
        }
        case 2:
        {
            coonfd_event |= EPOLLET;
            break;
        }
        case 3:
        {
            listen_event |=EPOLLET;
            coonfd_event |=EPOLLET;
            break;
        }
        default:
            break;
    } 
    Httpconn::isET_ = (coonfd_event && EPOLLET);
}
void process_poll::run_parent()
{
    setup_sig_pipe();
    _proc_status = true;
    //epoller_->AddFd_(sig_pipe[0],pipe_event);

    int events_number=0;
    int new_accept = 1;
    int epoll_time = -1;
    int proc_pos = 0;
    while(_proc_status ==true)
    {
        events_number = epoller_->Wait_(epoll_time);
        if(events_number == 0 && errno!=EINTR)
        {   /*std::cout<<"Time out"<<std::endl;*/ continue; }
        if( (events_number <0) && (errno != EINTR) )
        {   /*std::cout<<"epoll_wait Failed"<<std::endl;*/ perror("epoll_wait"); break; }
        for(int i=0;i<events_number;i++)
        {
            int Fd = epoller_->GetEventFd_(i);
            int Event = epoller_->GetEvent_(i);
            //处理 链接请求
            if( Fd == ListenFd_)
            {
                //DealListen();
                int p_index = proc_pos;
                do
                {//Round Rubin
                    p_index = (p_index+1)%_NOW_PROCES_NUMBER;
                } while (_proc_atribute[p_index].self_id == -1);    //循环队列
                proc_pos = p_index;
                send(_proc_atribute[p_index].pipe[0],(char*)&new_accept,sizeof(new_accept),0);
                //std::cout<<"轮转: "<<p_index<<std::endl;
            }
            //处理  信号
            //epoll_wait 第三个参数会返回 带有 因什么触发epoll的宏 
            else if( (Fd == sig_pipe[0]) && (Event & EPOLLIN) )
            {
                char recv_sig[1024];
                int recv_status = recv(Fd,recv_sig,sizeof(recv_sig),0);
                if( recv_status <=0 )
                {   /*std::cout<<"recv Faied OR no msg ..."<<std::endl;*/ perror("errno"); continue; }
                else
                {
                    for(int i = 0 ; i < recv_status ;i++)
                    {   switch(recv_sig[i])
                        {
                        case SIGINT:    
                        {
                            //父进程 被终止
                            pid_t pid;
                            std::cout<<"parent process: "<<getpid()<<" be killed ..."<<std::endl;
                            for(int i = 0 ; i <_NOW_PROCES_NUMBER;i++)
                            {
                                pid = _proc_atribute[i].self_id;
                                if( pid != -1 )
                                {
                                    kill(pid,SIGTERM);
                                }
                            }
                            std::cout<<"all process has killed ..."<<std::endl;
                            break;
                        }
                        case SIGCHLD:
                        {
                            //有子进程 结束 找出该子进程 并关闭资源
                            pid_t pid;
                            int exit_status;
                            while( ( pid = waitpid(-1,&exit_status,WNOHANG) ) >0)
                            {
                                for(int i = 0 ; i <_NOW_PROCES_NUMBER;i++)
                                {
                                    if(pid == _proc_atribute[i].self_id)
                                    {
                                        std::cout<<"p: child process: "<<pid<<" exit ... status: "<<exit_status<<std::endl;
                                       // perror("errno");
                                        close(_proc_atribute[i].pipe[0]);
                                        _proc_atribute[i].self_id = -1;
                                    }
                                }
                            }
                            //如果所有的子进程都结束了
                            //就结束父进程
                            _proc_status = false;
                            for(int i =0;i<_NOW_PROCES_NUMBER;i++)
                            {
                                if(_proc_atribute[i].self_id !=-1)
                                {   _proc_status = true; /*break;*/}
                            }
                            break;
                        }
                        case SIGTERM:
                        case SIGPIPE:   
                        default:
                        {
                        break;
                        }
                        }
                    }
                }
            }
            else
            {
                continue;
            }
        }
    }
    delete []_proc_atribute;
    //close(sig_pipe[0]);
    //close(sig_pipe[1]);
}
void process_poll::run_child()
{
    Timer_.reset(new HeapTime());
    epoller_.reset(new Epoller());

    srcDir_=getcwd(nullptr,256);
    assert(srcDir_);
    strcat(srcDir_,"/resource");
    Httpconn::srcDir_ = srcDir_;
    Httpconn::usercount_ = 0;

    setup_sig_pipe();
    //先找到本进程与父进程通信 的管道
    int self_pipe = _proc_atribute[_proc_index].pipe[1];
    epoller_->AddFd_(self_pipe,pipe_event);

    int events_number = -1;
    int recv_status= -1;
    int epoll_time = -1;
    //int count = 0;
    _proc_status = true;
    while( _proc_status == true )
    {        
        if(timeoutMS_ >0 )
        {
            epoll_time = Timer_->GetNetTick();            
        }
        events_number = epoller_->Wait_(timeoutMS_);
        for(int i = 0 ; i < events_number;i++)
        {
            int Fd = epoller_->GetEventFd_(i);
            int Event = epoller_->GetEvent_(i);
            //处理父进程 通过 通信管道 发送来的信息 准备接收客户端请求
            if( (Fd == self_pipe) && (Event & EPOLLIN))
            {
                int Client = -1;
                recv_status = recv(Fd,(char*)&Client,sizeof(Client),0);
                if( ( (recv_status <0 ) && ( errno != EAGAIN) ) || recv_status ==0 )
                {   continue;   }
                else
                {
                    //处理链接请求
                    DealAccept();
                }
                continue;
            }
            else if(Fd !=sig_pipe[0])
            {
                if(Event & ( EPOLLRDHUP | EPOLLHUP | EPOLLERR))
                {
                    assert(user_.count(Fd) > 0);
                    CloseCoon(&user_[Fd]);
                    continue;
                }
                else if(Event & EPOLLIN) 
                {   
                    //解析 请求报文
                    assert((user_.count(Fd) > 0));
                    DealRead(&user_[Fd]);
                    continue;
                }
                else if(Event & EPOLLOUT)
                {
                    //生成 回应报文
                    assert((user_.count(Fd)>0 ));
                    DealWrite(&user_[Fd]);
                    continue;
                }
            }
            //处理系统信号
            else if(Fd == sig_pipe[0]&& (Event& EPOLLIN ))
            {
                //处理信号
                char recv_sig[1024];
                int recv_status = recv(sig_pipe[0],recv_sig,sizeof(recv_sig),0);
                if( recv_status <=0 )
                {   std::cout<<"recv Failed OR no msg ..."<<std::endl; perror("recv"); continue; }
                else
                {
                    for(int i = 0 ; i< recv_status;i++)
                    {
                        switch(recv_sig[i])
                        {
                            case SIGINT:
                            {
                                std::cout<<"child process: "<<getpid()<<" SIGINT"<<std::endl;
                                _proc_status = false;
                                break;
                            }
                            case SIGCHLD:
                            {
                                pid_t pid;
                                int exit_status;
                                std::cout<<"child process SIGCHILD"<<std::endl;
                                while( (pid = waitpid(-1,&exit_status,WNOHANG) ) >0 )
                                {
                                    perror("errno");
                                    std::cout<<"child process: "<<pid<<"exit ...status: "<<exit_status<<std::endl;
                                    continue;
                                }
                                break;
                            }
                            case SIGTERM:
                            case SIGPIPE:
                            default:
                            {    
                                break;
                            }
                        }
                    }
                }
                continue;
            }
            else
            {
                std::cout<<"Unexpected event"<<std::endl;
                perror("errno");
                continue;;
            }
        }
    }
    close(self_pipe);
    delete [] _proc_atribute;
    //close(sig_pipe[0]);
    //close(sig_pipe[1]);
}
void process_poll::SendError(int fd,const char* info)
{
    assert(fd > 0);
    int status = send(fd,info,strlen(info),0);
    if(status < 0)
    {
        LOG_WARN("Send Error To Client[%d] Error",fd);
    }
    close(fd);
}
void process_poll::ExtenTime(Httpconn* Client)
{
    assert(Client != nullptr);
    if(timeoutMS_ >0 )
    {
        Timer_->adjust(Client->getFd(),timeoutMS_);
    }
}
void process_poll::CloseCoon(Httpconn* Client)
{
    assert(Client != nullptr);
    epoller_->DelFd_(Client->getFd());
    Client->Close();    
}
void process_poll::DealWrite(Httpconn* Client)
{
    assert(Client != nullptr);
    ExtenTime(Client);
    OnWrite(Client);
}
void process_poll::DealRead(Httpconn* Client)
{
    assert(Client != nullptr);
    ExtenTime(Client);
    Onread(Client);
}
void process_poll::Onread(Httpconn* Client)
{
    assert(Client != nullptr);
    int status = -1;
    int readError = 0;
    status = Client->read(&readError);
    if(status <=0 && readError!= EAGAIN)
    {
        CloseCoon(Client);
        return;
    }
    OnProcess(Client);
}
void process_poll::OnWrite(Httpconn* Client)
{
    std::unique_lock<std::mutex> lock(mtx);
    assert(Client != nullptr);
    int status = -1;
    int writeError = 0;
    status = Client->write(&writeError);
    if(Client->TotalWBytes() == 0)
    {//缓冲区的数据被全部聚集写入完毕
        if(Client->IsKeepAlive() == true)
        {
            epoller_->ModFd_(Client->getFd(),coonfd_event|EPOLLIN);
            return ;
        }
    }
    else if(status <=0 )
    {
        if(writeError == EAGAIN)    //缓冲区满,再次触发epoll_wait 写事件 + fd写事件
        {
            epoller_->ModFd_(Client->getFd(),coonfd_event|EPOLLOUT);
            return ;
        }
    }
    //至此,一次完整的事务/http请求 才完成
    //客户端的页面在一次请求中已经加载出了一次,无需再保持连接
    //关闭与客户端的连接
    CloseCoon(Client);
}
void process_poll::OnProcess(Httpconn* Client)
{
    assert(Client != nullptr);
    if(Client->process()==true)
    {//解析成功完成
    //将sock改为写预备 发送http回应报文
        epoller_->ModFd_(Client->getFd(),coonfd_event|EPOLLOUT);
    }
    else
    {//解析未完成,可能的原因是缓存区不够,再进行一轮读事件
        epoller_->ModFd_(Client->getFd(),coonfd_event|EPOLLIN);
    }
}

